package com.example.websocketserver.configuration.handler;

import com.alibaba.fastjson.JSONObject;
import com.example.websocketserver.configuration.WebsocketPushService;
import com.example.websocketserver.pojo.WebSocketConstants;
import com.example.websocketserver.pojo.WebSocketModuleEnum;
import com.example.websocketserver.pojo.WebsocketRealMessage;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author liulei
 * @version 1.0
 */
@Slf4j
//@Component TODO fix
@RequiredArgsConstructor
public class BusinessMessageReceiver {

    private final WebsocketPushService pushService;

    /**
     * 监听从其他平台过来的消息
     * 群发topic模板:
     * {"msgType":1,"module":"device","uniqueKey":"12312","content":"特定用户[{user_10000}]数据","receiver":null}
     *
     * 单点推送模板:
     * {"msgType":2,"module":"device","uniqueKey":"12312","content":"来自于topic的业务mq数据","receiver":"10000"}
     */
    @RabbitListener(queues = WebSocketConstants.BUSINESS_QUEUE)
    public void consume(@Payload Message message, @Header(AmqpHeaders.CHANNEL) Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        String jsonResult = new String(message.getBody());
        log.info("websocket->monitor business queue listen message|{}", jsonResult);
        try {
            WebsocketRealMessage socketMessage = JSONObject.parseObject(jsonResult, WebsocketRealMessage.class);
            Integer msgType = socketMessage.getMsgType();
            String module = socketMessage.getModule();
            String uniqueKey = socketMessage.getUniqueKey();
            String content = socketMessage.getContent();
            if (null == msgType || StringUtils.isEmpty(module) || StringUtils.isEmpty(uniqueKey) || StringUtils.isEmpty(content)) {
                log.error("websocket->monitor business queue listen message fail |miss key value");
                return;
            }
            WebSocketModuleEnum pushModule = WebSocketModuleEnum.getByModule(module);
            // TODO 待补全推送判断和字段
            // topic级别推送
            if (msgType == 1) {
                log.info("websocket->push topic param {}", socketMessage);
                pushService.push2Topic(pushModule.getTopicName(uniqueKey), socketMessage);
                return;
            }
            // 用户级别推送
            if (StringUtils.isEmpty(socketMessage.getReceiver())) {
                return;
            }
            pushService.push2User(WebSocketConstants.getUserId(socketMessage.getReceiver()), "/queue/point", socketMessage);
            log.info("websocket->push param succeed");
        } catch (Exception e) {
            log.error("websocket->monitor business queue listen message |error|{}", jsonResult, e);
        } finally {
            channel.basicAck(deliveryTag, true);
        }
    }
}
